package com.atguigu.dga.meta.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SimplePropertyPreFilter;
import com.atguigu.dga.common.utils.SqlUtil;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.bean.TableMetaInfoExtra;
import com.atguigu.dga.meta.bean.TableMetaInfoQuery;
import com.atguigu.dga.meta.bean.TableMetaInfoVO;
import com.atguigu.dga.meta.mapper.TableMetaInfoMapper;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.thrift.TException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;

/**
 * <p>
 * 元数据表 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2024-07-24
 */
@Service
@DS("dga")
public class TableMetaInfoServiceImpl extends ServiceImpl<TableMetaInfoMapper, TableMetaInfo> implements TableMetaInfoService {

    @Value("${hive.metastore.uris}")
    String hiveMetaUri;

    @Value("${default.database}")
    String schemaName;

    HiveMetaStoreClient hiveMetaStoreClient ;

    @Autowired
    TableMetaInfoExtraService tableMetaInfoExtraService;

    @PostConstruct
    public void  initHiveClient(){
        HiveConf hiveConf = new HiveConf();
        hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS,hiveMetaUri);
        try {
            hiveMetaStoreClient=  new HiveMetaStoreClient(hiveConf);
        } catch (MetaException e) {
            throw new RuntimeException(e);
        }
    }

    public void initTableMetaInfo(String assessDate) throws TException, URISyntaxException, IOException, InterruptedException {
        System.out.println("hiveMetaUri = " + hiveMetaUri);


        //  1 从hive中提取元数据列表  List<TableMetaInfo> tableMetaInfoList =  ....
        List<String> tableNameList = hiveMetaStoreClient.getAllTables(schemaName);
        List<TableMetaInfo> tableMetaInfoList =new LinkedList<>();
        for (String tableName : tableNameList) {

            // 根据表名库名 提取元数据 对象 TableMetaInfo
            TableMetaInfo tableMetaInfo = this.extractTableMetaInfoFromHive(tableName, schemaName);
            System.out.println("tableMetaInfo = " + tableMetaInfo);

            // 从hdfs 中提取信息
            this.extractHdfsInfo(tableMetaInfo);

            tableMetaInfo.setAssessDate(assessDate);
            tableMetaInfo.setCreateTime(new Date());
            tableMetaInfoList.add(tableMetaInfo);
        }
        //2 保存元数据列表
        this.remove(new QueryWrapper<TableMetaInfo>().eq("assess_date",assessDate));
         saveBatch(tableMetaInfoList);
        //3 初始化辅助信息 （只初始化不存在辅助信息的表)
        //3.1 筛选出没有辅助信息的元数据表
        //不推荐循环查询数据库
//        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
//            TableMetaInfoExtra infoExtra = tableMetaInfoExtraService.getOne(new QueryWrapper<TableMetaInfoExtra>().eq("table_name", tableMetaInfo.getTableName()).eq("schema_name", tableMetaInfo.getSchemaName()));
//
//        }
        QueryWrapper queryWrapper=  new QueryWrapper<TableMetaInfo>().eq("assess_date",assessDate)
                 .notInSql("concat(schema_name,table_name)", "select concat(schema_name,table_name) from table_meta_info_extra");
        List tableMetaListWithoutExists = this.list(queryWrapper);

        //补充辅助信息
        tableMetaInfoExtraService.initTableMetaInfoExtra(tableMetaListWithoutExists);


    }


    private TableMetaInfo extractTableMetaInfoFromHive(String tableName,String schemaName) throws TException {
        Table table = hiveMetaStoreClient.getTable(schemaName, tableName);


        TableMetaInfo tableMetaInfo = new TableMetaInfo();
        tableMetaInfo.setTableName(table.getTableName());
        tableMetaInfo.setSchemaName(schemaName);
        //创建时间
        Date createTime = new Date(table.getCreateTime() * 1000L);
        String createTimeStr = DateFormatUtils.format(createTime, "yyyy-MM-dd HH:mm:ss");
        tableMetaInfo.setTableCreateTime(createTimeStr);
        //所有者
        tableMetaInfo.setTableFsOwner(table.getOwner());
        //参数集合
        tableMetaInfo.setTableParametersJson(JSON.toJSONString(table.getParameters() ));
        //分区键
        SimplePropertyPreFilter colsFilter = new SimplePropertyPreFilter("comment", "name", "type");
        tableMetaInfo.setPartitionColNameJson(JSON.toJSONString(table.getPartitionKeys(),colsFilter));
        StorageDescriptor storageDescriptor = table.getSd();//提取存储描述器
        //普通字段
        tableMetaInfo.setColNameJson(JSON.toJSONString(storageDescriptor.getCols(),colsFilter));
        //输入输出格式
        tableMetaInfo.setTableInputFormat(storageDescriptor.getInputFormat());
        tableMetaInfo.setTableOutputFormat(storageDescriptor.getOutputFormat());
        //分桶字段
        tableMetaInfo.setTableBucketColsJson(JSON.toJSONString(storageDescriptor.getBucketCols()));
        //分桶排序列
        tableMetaInfo.setTableSortColsJson(JSON.toJSONString(storageDescriptor.getSortCols()));
        //分桶数
        tableMetaInfo.setTableBucketNum(storageDescriptor.getNumBuckets()+0L);
        //存储位置
        tableMetaInfo.setTableFsPath(storageDescriptor.getLocation());
        //序列化类
        tableMetaInfo.setTableRowFormatSerde(storageDescriptor.getSerdeInfo().getSerializationLib());
        //表类型
        tableMetaInfo.setTableType(table.getTableType());
        //表备注
        tableMetaInfo.setTableComment(table.getParameters().get("comment"));

        return tableMetaInfo;

    }

    //补充hdfs的相关信息
    //准备阶段
    //       1.1  递归的起点  （一级子目录集合）
    //       1.2  递归的访问工具 （hdfs路径的访问工具    FileSystem ）
    //       1.3  递归的存储结果容器  （TableMetaInfo）
    //
    //递归的执行阶段
    //         遍历所有的一级子目录
    //             2.1   叶子节点  （文件）
    //                        处理： 计算 汇总 比较 …. 处理结果放入容器
    //		      处理完  结束
    //             2.2   分支节点     (目录)
    //                         处理：有可能也需要处理  也可能不需要
    //                         展开其下的子节点 作为新的访问起点
    //                          调用本方法形成递归
    private void extractHdfsInfo(TableMetaInfo tableMetaInfo) throws URISyntaxException, IOException, InterruptedException {
        //table_size遍历 hdfs对应的表目录下的所有文件 加和
        //table_total_size遍历 hdfs对应的表目录下的所有文件 乘以副本 加和
        // table_last_modify_time   遍历 hdfs对应的表目录下的所有文件 修改时间取最大值
        //table_last_access_time  遍历 hdfs对应的表目录下的所有文件 访问时间取最大值

       // 1.1  递归的访问工具 （hdfs路径的访问工具    FileSystem ）
        URI uri = new URI(tableMetaInfo.getTableFsPath());
        FileSystem fileSystem = FileSystem.get(uri, new Configuration(), tableMetaInfo.getTableFsOwner());
        //  一级子节点
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(tableMetaInfo.getTableFsPath()));
        //1.3  递归的存储结果容器  （TableMetaInfo）
        getHdfsInfoRec(fileSystem,fileStatuses,tableMetaInfo);

        System.out.println("tableMetaInfo =  " +tableMetaInfo.getTableName()+":"+ tableMetaInfo.getTableSize()
                +":accessTime:"+tableMetaInfo.getTableLastAccessTime()
                +"modifyTIme:"+tableMetaInfo.getTableLastModifyTime());
        //fs_capcity_size 取当前hdfs环境信息
        //fs_remain_size 取当前hdfs环境信息
        //fs_used_size  取当前hdfs环境信息
        tableMetaInfo.setFsCapcitySize(fileSystem.getStatus().getCapacity());
        tableMetaInfo.setFsRemainSize(fileSystem.getStatus().getRemaining());
        tableMetaInfo.setFsUsedSize(fileSystem.getStatus().getUsed());

    }

    //递归的执行阶段
    //         遍历所有的一级子目录
    //             2.1   叶子节点  （文件）
    //                        处理： 计算 汇总 比较 …. 处理结果放入容器
    //		      处理完  结束
    //             2.2   分支节点     (目录)
    //                         处理：有可能也需要处理  也可能不需要
    //                         展开其下的子节点 作为新的访问起点
    //                          调用本方法形成递归
    private void getHdfsInfoRec(FileSystem fileSystem, FileStatus[] fileStatuses, TableMetaInfo tableMetaInfo) throws IOException {
        for (FileStatus fileStatus : fileStatuses) { //    遍历所有的一级子目录
           // 2.1   叶子节点  （文件）
            if(fileStatus.isFile()){
                tableMetaInfo.setTableSize(tableMetaInfo.getTableSize() + fileStatus.getLen());
                tableMetaInfo.setTableTotalSize(  tableMetaInfo.getTableTotalSize() + fileStatus.getLen()*fileStatus.getReplication());
                //访问时间处理 访问时间取最大值

                //如果没有时间 直接填入
                //如果有时间  进行比较 保留最大值
                if(tableMetaInfo.getTableLastAccessTime()==null){
                    tableMetaInfo.setTableLastAccessTime( new Date(fileStatus.getAccessTime()));
                }else{
                    if(tableMetaInfo.getTableLastAccessTime().getTime()<fileStatus.getAccessTime()){
                        tableMetaInfo.setTableLastAccessTime( new Date(fileStatus.getAccessTime()));
                    }
                }

                if(tableMetaInfo.getTableLastModifyTime()==null){
                    tableMetaInfo.setTableLastModifyTime( new Date(fileStatus.getModificationTime()));
                }else{
                    if(tableMetaInfo.getTableLastModifyTime().getTime()<fileStatus.getModificationTime()){
                        tableMetaInfo.setTableLastModifyTime( new Date(fileStatus.getModificationTime()));
                    }
                }


            }else{
                //             2.2   分支节点     (目录)
                //                         处理：有可能也需要处理  也可能不需要
                //                         展开其下的子节点 作为新的访问起点
                FileStatus[] subFileStatuses = fileSystem.listStatus(fileStatus.getPath());
                //                          调用本方法形成递归
                this.getHdfsInfoRec(fileSystem,subFileStatuses,tableMetaInfo);
            }
        }



    }




    @Override
    public List<TableMetaInfoVO> getListForQuery(TableMetaInfoQuery tableMetaInfoQuery) {
        //实现 table_meta_info 和 table_meta_info_extra 的关联查询
        StringBuilder sb =new StringBuilder(2048);

        sb.append(  "select    tm.id ,tm.table_name,tm.schema_name,table_comment,table_size,table_total_size,tec_owner_user_name,\n" +
                "     busi_owner_user_name, table_last_access_time,table_last_modify_time from table_meta_info  tm\n" +
                "     left  join table_meta_info_extra  te\n" +
                "     on tm.table_name=te.table_name  and tm.schema_name=te.schema_name\n" +
                "where assess_date = (select max( assess_date ) from table_meta_info tmi\n" +
                "     where tm.table_name = tmi.table_name\n" +
                "     and tm.schema_name=tmi.schema_name  )");


        //接收动态条件

        if(tableMetaInfoQuery.getSchemaName()!=null&&!tableMetaInfoQuery.getSchemaName().trim().equals("")){
            sb.append(" and tm.schema_name like '%"+ SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getSchemaName()) +"%'");
        }
        if (tableMetaInfoQuery.getTableName()!=null&&!tableMetaInfoQuery.getTableName().trim().equals("")){
            sb.append(" and tm.table_name like '%"+SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getTableName())+"%'");

        }
        if (tableMetaInfoQuery.getDwLevel()!=null&&!tableMetaInfoQuery.getDwLevel().trim().equals("")){
            sb.append(" and te.dw_level like '%"+SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getDwLevel())+"%'");
        }

        //处理分页
        // 行号= （页码-1）* 每页行数
        // limit 行号，每页行数
        sb.append(" limit "+(tableMetaInfoQuery.getPageNo()-1)*tableMetaInfoQuery.getPageSize()+","+tableMetaInfoQuery.getPageSize());


        List<TableMetaInfoVO> tableMetaInfoVOList = baseMapper.selectTableMetaListForQuery(sb.toString());

        return tableMetaInfoVOList;
    }

    @Override
    public Integer getTotalForQuery(TableMetaInfoQuery tableMetaInfoQuery) {
        //实现 table_meta_info 和 table_meta_info_extra 的关联查询
        StringBuilder sb =new StringBuilder(2048);

        sb.append(  "select  count(*)     from table_meta_info  tm\n" +
                "     left  join table_meta_info_extra  te\n" +
                "     on tm.table_name=te.table_name  and tm.schema_name=te.schema_name\n" +
                "where assess_date = (select max( assess_date ) from table_meta_info tmi\n" +
                "     where tm.table_name = tmi.table_name\n" +
                "     and tm.schema_name=tmi.schema_name  )");


        //接收动态条件

        if(tableMetaInfoQuery.getSchemaName()!=null&&!tableMetaInfoQuery.getSchemaName().trim().equals("")){
            sb.append(" and tm.schema_name like '%"+ SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getSchemaName()) +"%'");
        }
        if (tableMetaInfoQuery.getTableName()!=null&&!tableMetaInfoQuery.getTableName().trim().equals("")){
            sb.append(" and tm.table_name like '%"+SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getTableName())+"%'");

        }
        if (tableMetaInfoQuery.getDwLevel()!=null&&!tableMetaInfoQuery.getDwLevel().trim().equals("")){
            sb.append(" and te.dw_level like '%"+SqlUtil.filterUnsafeSql(tableMetaInfoQuery.getDwLevel())+"%'");
        }

        //处理分页
        // 行号= （页码-1）* 每页行数
        // limit 行号，每页行数

        Integer total= baseMapper.selectTableMetaTotalForQuery(sb.toString());

        return total;
    }

    @Override
    public List<TableMetaInfo> getListWithExtra(String assessDate) {
        List<TableMetaInfo> tableMetaInfoList = baseMapper.selectListWithExtra(assessDate);
        return tableMetaInfoList;
    }
}
